package com.njbdqn.call

import com.njbdqn.util.{HdfsConnection, MysqlConnection}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

/**
 * 全局热卖召回 将数据处理存放到hdfs
 */
object GlobalHotSellCall {
  def hotSell(spark:SparkSession)={
    val oitab = MysqlConnection.readTable(spark,"orderItems")
    // 计算全局热卖商品前100名
    val tjTop100 = oitab.groupBy("good_id").agg(sum("buy_num").as("sellnum")).orderBy(desc("sellnum")).limit(100)
    // 将所有用户的id和推荐前100名笛卡尔积
    //找到所有用户
    val users = MysqlConnection.readTable(spark,"customs").select("cust_id")
    // 最终全局推荐
    val res = users.crossJoin(tjTop100)
    // 并把数据存放到hdfs
    HdfsConnection.writeDataToHdfs("/kb08/myshops/dwd_hotsell",res)
    // 并且将热卖召回的前10名存放到mysql的hotsell
    val hotSell = tjTop100.limit(10)
    MysqlConnection.writeTable(spark,hotSell,"hotsell")
  }
}
